package com.tico.editor.editors.kafkaThread;

import com.tico.editor.editors.domain.Kafka;
import com.tico.editor.editors.domain.TaskScheduled;
import com.tico.editor.editors.service.HttpEditorService;
import com.tico.editor.editors.service.KafkaService;
import com.tico.editor.editors.service.KafkaTopicService;
import com.tico.editor.editors.service.TaskScheduledService;
import com.tico.editor.editors.task.MyTask;
import com.tico.editor.editors.task.ScheduleTask;
import com.tico.editor.editors.task.ScheduleUtil;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentLinkedQueue;

public class TaskThread implements Runnable{
    public static final Map<String,ScheduleTask> scheduleTaskMap = new HashMap<>();

    private KafkaService kafkaService;
    private KafkaTopicService kafkaTopicService;
    private TaskScheduledService scheduledService;
    private HttpEditorService httpEditorService;

    public TaskThread(KafkaService kafkaService, KafkaTopicService kafkaTopicService, TaskScheduledService scheduledService, HttpEditorService httpEditorService) {
        this.kafkaService = kafkaService;
        this.kafkaTopicService = kafkaTopicService;
        this.scheduledService = scheduledService;
        this.httpEditorService = httpEditorService;
    }
    @Override
    public void run() {
        while (true){
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            HashMap<String,Object> param = new HashMap<>();
            param.put("taskType","scheduled");
            List<TaskScheduled> taskScheduleds = scheduledService.findByNamedParamList(param);
            System.out.println("任务列表---------------------"+taskScheduleds.size());
            if(taskScheduleds.size()==0){
                return;
            }
            for(TaskScheduled taskScheduled:taskScheduleds){
                if(scheduleTaskMap!=null){
                    if(taskScheduled.getStatus().equals("false")){
                        ScheduleTask scheduleTask = scheduleTaskMap.get(taskScheduled.getId());
                        try {
                            if(scheduleTask!=null){
                                ScheduleUtil.cancel(scheduleTask);
                            }
                        }catch (Exception e){
                            e.printStackTrace();
                        }
                    }else{
                        ScheduleTask scheduleTask = new MyTask(taskScheduled.getId(),taskScheduled.getId(),taskScheduled,httpEditorService);//new一个具体的执行任务
                        ScheduleUtil.start(scheduleTask, Integer.valueOf(taskScheduled.getTaskScheduled()));//每多少秒执行一次
                        scheduleTaskMap.put(taskScheduled.getId(),scheduleTask);
                    }
                }
            }
        }
    }
}
